/*
 * contrib/pgstattuple/pgstattuple.c
 *
 * Copyright (c) 2001,2002	Tatsuo Ishii
 *
 * Permission to use, copy, modify, and distribute this software and
 * its documentation for any purpose, without fee, and without a
 * written agreement is hereby granted, provided that the above
 * copyright notice and this paragraph and the following two
 * paragraphs appear in all copies.
 *
 * IN NO EVENT SHALL THE AUTHOR BE LIABLE TO ANY PARTY FOR DIRECT,
 * INDIRECT, SPECIAL, INCIDENTAL, OR CONSEQUENTIAL DAMAGES, INCLUDING
 * LOST PROFITS, ARISING OUT OF THE USE OF THIS SOFTWARE AND ITS
 * DOCUMENTATION, EVEN IF THE UNIVERSITY OF CALIFORNIA HAS BEEN ADVISED
 * OF THE POSSIBILITY OF SUCH DAMAGE.
 *
 * THE AUTHOR SPECIFICALLY DISCLAIMS ANY WARRANTIES, INCLUDING, BUT NOT
 * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
 * A PARTICULAR PURPOSE.  THE SOFTWARE PROVIDED HEREUNDER IS ON AN "AS
 * IS" BASIS, AND THE AUTHOR HAS NO OBLIGATIONS TO PROVIDE MAINTENANCE,
 * SUPPORT, UPDATES, ENHANCEMENTS, OR MODIFICATIONS.
 */

#include "postgres.h"

#include "access/gist_private.h"
#include "access/hash.h"
#include "access/heapam.h"
#include "access/nbtree.h"
#include "access/relscan.h"
#include "access/tableam.h"
#include "catalog/namespace.h"
#include "catalog/pg_am_d.h"
#include "funcapi.h"
#include "miscadmin.h"
#include "storage/bufmgr.h"
#include "storage/lmgr.h"
#include "utils/builtins.h"
#include "utils/varlena.h"

PG_MODULE_MAGIC;

PG_FUNCTION_INFO_V1(pgstattuple);
PG_FUNCTION_INFO_V1(pgstattuple_v1_5);
PG_FUNCTION_INFO_V1(pgstattuplebyid);
PG_FUNCTION_INFO_V1(pgstattuplebyid_v1_5);

/*
 * struct pgstattuple_type
 *
 * tuple_percent、dead_tuple_percent和free_percent是可计算的，
 * 因此在此未定义。
 */
typedef struct pgstattuple_type
{
	uint64		table_len;
	uint64		tuple_count;
	uint64		tuple_len;
	uint64		dead_tuple_count;
	uint64		dead_tuple_len;
	uint64		free_space;		/* 可用/可重用空间（以字节为单位） */
} pgstattuple_type;

typedef void (*pgstat_page) (pgstattuple_type *, Relation, BlockNumber,
							 BufferAccessStrategy);

static Datum fc_build_pgstattuple_type(pgstattuple_type *fc_stat,
									FunctionCallInfo fc_fcinfo);
static Datum fc_pgstat_relation(Relation fc_rel, FunctionCallInfo fc_fcinfo);
static Datum fc_pgstat_heap(Relation fc_rel, FunctionCallInfo fc_fcinfo);
static void fc_pgstat_btree_page(pgstattuple_type *fc_stat,
							  Relation fc_rel, BlockNumber fc_blkno,
							  BufferAccessStrategy fc_bstrategy);
static void fc_pgstat_hash_page(pgstattuple_type *fc_stat,
							 Relation fc_rel, BlockNumber fc_blkno,
							 BufferAccessStrategy fc_bstrategy);
static void fc_pgstat_gist_page(pgstattuple_type *fc_stat,
							 Relation fc_rel, BlockNumber fc_blkno,
							 BufferAccessStrategy fc_bstrategy);
static Datum fc_pgstat_index(Relation fc_rel, BlockNumber fc_start,
						  pgstat_page fc_pagefn, FunctionCallInfo fc_fcinfo);
static void fc_pgstat_index_page(pgstattuple_type *fc_stat, Page fc_page,
							  OffsetNumber fc_minoff, OffsetNumber fc_maxoff);

/*
 * build_pgstattuple_type -- 构建pgstattuple_type元组
 */
static Datum fc_build_pgstattuple_type(pgstattuple_type *fc_stat, FunctionCallInfo fc_fcinfo)
{
#define NCOLUMNS	9
#define NCHARS		314

	HeapTuple	fc_tuple;
	char	   *fc_values[NCOLUMNS];
	char		fc_values_buf[NCOLUMNS][NCHARS];
	int			fc_i;
	double		fc_tuple_percent;
	double		fc_dead_tuple_percent;
	double		fc_free_percent;	/* 可用/可重用空间（以百分比表示） */
	TupleDesc	fc_tupdesc;
	AttInMetadata *fc_attinmeta;

	/* 为我们的结果类型构建一个元组描述符 */
	if (get_call_result_type(fc_fcinfo, NULL, &fc_tupdesc) != TYPEFUNC_COMPOSITE)
		elog(ERROR, "return type must be a row type");

	/*
	 * 生成稍后从原始C字符串生成元组所需的属性元数据
	 */
	fc_attinmeta = TupleDescGetAttInMetadata(fc_tupdesc);

	if (fc_stat->table_len == 0)
	{
		fc_tuple_percent = 0.0;
		fc_dead_tuple_percent = 0.0;
		fc_free_percent = 0.0;
	}
	else
	{
		fc_tuple_percent = 100.0 * fc_stat->tuple_len / fc_stat->table_len;
		fc_dead_tuple_percent = 100.0 * fc_stat->dead_tuple_len / fc_stat->table_len;
		fc_free_percent = 100.0 * fc_stat->free_space / fc_stat->table_len;
	}

	/*
	 * 准备一个值数组用于构造元组。这个数组应该是C字符串数组，
	 * 稍后将由适当的“in”函数处理。
	 */
	for (fc_i = 0; fc_i < NCOLUMNS; fc_i++)
		fc_values[fc_i] = fc_values_buf[fc_i];
	fc_i = 0;
	snprintf(fc_values[fc_i++], NCHARS, INT64_FORMAT, fc_stat->table_len);
	snprintf(fc_values[fc_i++], NCHARS, INT64_FORMAT, fc_stat->tuple_count);
	snprintf(fc_values[fc_i++], NCHARS, INT64_FORMAT, fc_stat->tuple_len);
	snprintf(fc_values[fc_i++], NCHARS, "%.2f", fc_tuple_percent);
	snprintf(fc_values[fc_i++], NCHARS, INT64_FORMAT, fc_stat->dead_tuple_count);
	snprintf(fc_values[fc_i++], NCHARS, INT64_FORMAT, fc_stat->dead_tuple_len);
	snprintf(fc_values[fc_i++], NCHARS, "%.2f", fc_dead_tuple_percent);
	snprintf(fc_values[fc_i++], NCHARS, INT64_FORMAT, fc_stat->free_space);
	snprintf(fc_values[fc_i++], NCHARS, "%.2f", fc_free_percent);

	/* 构建一个元组 */ 
	fc_tuple = BuildTupleFromCStrings(fc_attinmeta, fc_values);

	/* 将元组转化为一个 datum */
	return HeapTupleGetDatum(fc_tuple);
}

/* ----------
 * pgstattuple:
 * 返回活/死元组信息
 *
 * C FUNCTION定义
 * pgstattuple(text) 返回 pgstattuple_type
 *
 * 此处的superuser()检查必须保留，因为库可能会升级，
 * 而扩展未升级，这意味着在1.5版本之前的安装中
 * 这些函数可能会被任何用户调用。
 * ----------
 */

Datum pgstattuple(PG_FUNCTION_ARGS)
{
	text	   *fc_relname = PG_GETARG_TEXT_PP(0);
	RangeVar   *fc_relrv;
	Relation	fc_rel;

	if (!superuser())
		ereport(ERROR,
				(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
				 errmsg("must be superuser to use pgstattuple functions")));

	/* 打开关系 */
	fc_relrv = makeRangeVarFromNameList(textToQualifiedNameList(fc_relname));
	fc_rel = relation_openrv(fc_relrv, AccessShareLock);

	PG_RETURN_DATUM(fc_pgstat_relation(fc_rel, fcinfo));
}

/*
 * 从pgstattuple版本1.5开始，我们不再需要检查用户
 * 是否为超级用户，因为我们从PUBLIC中撤销
 * 函数的执行权限。
 * 用户可以根据其策略授予访问权限。
 *
 * 在其他方面与pgstattuple（上述）相同。
 */
Datum pgstattuple_v1_5(PG_FUNCTION_ARGS)
{
	text	   *fc_relname = PG_GETARG_TEXT_PP(0);
	RangeVar   *fc_relrv;
	Relation	fc_rel;

	/* 打开关系 */
	fc_relrv = makeRangeVarFromNameList(textToQualifiedNameList(fc_relname));
	fc_rel = relation_openrv(fc_relrv, AccessShareLock);

	PG_RETURN_DATUM(fc_pgstat_relation(fc_rel, fcinfo));
}

/* 必须保持 superuser() 检查，见上文。 */
Datum pgstattuplebyid(PG_FUNCTION_ARGS)
{
	Oid			fc_relid = PG_GETARG_OID(0);
	Relation	fc_rel;

	if (!superuser())
		ereport(ERROR,
				(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
				 errmsg("must be superuser to use pgstattuple functions")));

	/* 打开关系 */
	fc_rel = relation_open(fc_relid, AccessShareLock);

	PG_RETURN_DATUM(fc_pgstat_relation(fc_rel, fcinfo));
}

/* 移除1.5版本的superuser()检查，见上文 */
Datum pgstattuplebyid_v1_5(PG_FUNCTION_ARGS)
{
	Oid			fc_relid = PG_GETARG_OID(0);
	Relation	fc_rel;

	/* 打开关系 */
	fc_rel = relation_open(fc_relid, AccessShareLock);

	PG_RETURN_DATUM(fc_pgstat_relation(fc_rel, fcinfo));
}

/*
 * pgstat_relation
 */
static Datum fc_pgstat_relation(Relation fc_rel, FunctionCallInfo fc_fcinfo)
{
	const char *fc_err;

	/*
	 * 拒绝尝试读取非本地临时关系；我们可能会获取错误的数据，因为
	 * 我们无法查看拥有会话的本地缓冲区。
	 */
	if (RELATION_IS_OTHER_TEMP(fc_rel))
		ereport(ERROR,
				(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
				 errmsg("cannot access temporary tables of other sessions")));

	if (RELKIND_HAS_TABLE_AM(fc_rel->rd_rel->relkind) ||
		fc_rel->rd_rel->relkind == RELKIND_SEQUENCE)
	{
		return fc_pgstat_heap(fc_rel, fc_fcinfo);
	}
	else if (fc_rel->rd_rel->relkind == RELKIND_INDEX)
	{
		/* 见 pgstatindex_impl */
		if (!fc_rel->rd_index->indisvalid)
			ereport(ERROR,
					(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
					 errmsg("index \"%s\" is not valid",
							RelationGetRelationName(fc_rel))));

		switch (fc_rel->rd_rel->relam)
		{
			case BTREE_AM_OID:
				return fc_pgstat_index(fc_rel, BTREE_METAPAGE + 1,
									fc_pgstat_btree_page, fc_fcinfo);
			case HASH_AM_OID:
				return fc_pgstat_index(fc_rel, HASH_METAPAGE + 1,
									fc_pgstat_hash_page, fc_fcinfo);
			case GIST_AM_OID:
				return fc_pgstat_index(fc_rel, GIST_ROOT_BLKNO + 1,
									fc_pgstat_gist_page, fc_fcinfo);
			case GIN_AM_OID:
				fc_err = "gin index";
				break;
			case SPGIST_AM_OID:
				fc_err = "spgist index";
				break;
			case BRIN_AM_OID:
				fc_err = "brin index";
				break;
			default:
				fc_err = "unknown index";
				break;
		}
		ereport(ERROR,
				(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
				 errmsg("index \"%s\" (%s) is not supported",
						RelationGetRelationName(fc_rel), fc_err)));
	}
	else
	{
		ereport(ERROR,
				(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
				 errmsg("cannot get tuple-level statistics for relation \"%s\"",
						RelationGetRelationName(fc_rel)),
				 errdetail_relkind_not_supported(fc_rel->rd_rel->relkind)));
	}

	return 0;					/* 不应该发生 */
}

/*
 * pgstat_heap -- 返回堆中活/死元组信息
 */
static Datum fc_pgstat_heap(Relation fc_rel, FunctionCallInfo fc_fcinfo)
{
	TableScanDesc fc_scan;
	HeapScanDesc fc_hscan;
	HeapTuple	fc_tuple;
	BlockNumber fc_nblocks;
	BlockNumber fc_block = 0;		/* 下一个块以计数自由空间 */
	BlockNumber fc_tupblock;
	Buffer		fc_buffer;
	pgstattuple_type fc_stat = {0};
	SnapshotData fc_SnapshotDirty;

	/*
	 * 序列始终使用堆 AM，但它们在目录中并未显示这一点。
	 */
	if (fc_rel->rd_rel->relkind != RELKIND_SEQUENCE &&
		fc_rel->rd_rel->relam != HEAP_TABLE_AM_OID)
		ereport(ERROR,
				(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
				 errmsg("only heap AM is supported")));

	/* 禁用syncscan，因为我们假设从块零开始向上扫描 */
	fc_scan = table_beginscan_strat(fc_rel, SnapshotAny, 0, NULL, true, false);
	fc_hscan = (HeapScanDesc) fc_scan;

	InitDirtySnapshot(fc_SnapshotDirty);

	fc_nblocks = fc_hscan->rs_nblocks;	/* 要扫描的块数 */

	/* 扫描关系 */
	while ((fc_tuple = heap_getnext(fc_scan, ForwardScanDirection)) != NULL)
	{
		CHECK_FOR_INTERRUPTS();

		/* 必须持有缓冲区锁才能调用HeapTupleSatisfiesVisibility */
		LockBuffer(fc_hscan->rs_cbuf, BUFFER_LOCK_SHARE);

		if (HeapTupleSatisfiesVisibility(fc_tuple, &fc_SnapshotDirty, fc_hscan->rs_cbuf))
		{
			fc_stat.tuple_len += fc_tuple->t_len;
			fc_stat.tuple_count++;
		}
		else
		{
			fc_stat.dead_tuple_len += fc_tuple->t_len;
			fc_stat.dead_tuple_count++;
		}

		LockBuffer(fc_hscan->rs_cbuf, BUFFER_LOCK_UNLOCK);

		/*
		 * 为了避免物理读取表两次，尝试并行进行
		 * 自由空间扫描与堆扫描。然而，
		 * heap_getnext可能在给定页面上找不到元组，
		 * 所以我们不能简单地检查堆扫描返回的页面。
		 */
		fc_tupblock = ItemPointerGetBlockNumber(&fc_tuple->t_self);

		while (fc_block <= fc_tupblock)
		{
			CHECK_FOR_INTERRUPTS();

			fc_buffer = ReadBufferExtended(fc_rel, MAIN_FORKNUM, fc_block,
										RBM_NORMAL, fc_hscan->rs_strategy);
			LockBuffer(fc_buffer, BUFFER_LOCK_SHARE);
			fc_stat.free_space += PageGetHeapFreeSpace((Page) BufferGetPage(fc_buffer));
			UnlockReleaseBuffer(fc_buffer);
			fc_block++;
		}
	}

	while (fc_block < fc_nblocks)
	{
		CHECK_FOR_INTERRUPTS();

		fc_buffer = ReadBufferExtended(fc_rel, MAIN_FORKNUM, fc_block,
									RBM_NORMAL, fc_hscan->rs_strategy);
		LockBuffer(fc_buffer, BUFFER_LOCK_SHARE);
		fc_stat.free_space += PageGetHeapFreeSpace((Page) BufferGetPage(fc_buffer));
		UnlockReleaseBuffer(fc_buffer);
		fc_block++;
	}

	table_endscan(fc_scan);
	relation_close(fc_rel, AccessShareLock);

	fc_stat.table_len = (uint64) fc_nblocks * BLCKSZ;

	return fc_build_pgstattuple_type(&fc_stat, fc_fcinfo);
}

/*
 * pgstat_btree_page -- 检查btree页面中的元组
 */
static void fc_pgstat_btree_page(pgstattuple_type *fc_stat, Relation fc_rel, BlockNumber fc_blkno,
				  BufferAccessStrategy fc_bstrategy)
{
	Buffer		fc_buf;
	Page		fc_page;

	fc_buf = ReadBufferExtended(fc_rel, MAIN_FORKNUM, fc_blkno, RBM_NORMAL, fc_bstrategy);
	LockBuffer(fc_buf, BT_READ);
	fc_page = BufferGetPage(fc_buf);

	/* 页面有效，查看处理方式 */
	if (PageIsNew(fc_page))
	{
		/* 完全空页面 */
		fc_stat->free_space += BLCKSZ;
	}
	else
	{
		BTPageOpaque fc_opaque;

		fc_opaque = BTPageGetOpaque(fc_page);
		if (P_IGNORE(fc_opaque))
		{
			/* 删除或半死页面 */
			fc_stat->free_space += BLCKSZ;
		}
		else if (P_ISLEAF(fc_opaque))
		{
			fc_pgstat_index_page(fc_stat, fc_page, P_FIRSTDATAKEY(fc_opaque),
							  PageGetMaxOffsetNumber(fc_page));
		}
		else
		{
			/* 内部页面 */
		}
	}

	_bt_relbuf(fc_rel, fc_buf);
}

/*
 * pgstat_hash_page -- 检查哈希页面中的元组
 */
static void fc_pgstat_hash_page(pgstattuple_type *fc_stat, Relation fc_rel, BlockNumber fc_blkno,
				 BufferAccessStrategy fc_bstrategy)
{
	Buffer		fc_buf;
	Page		fc_page;

	fc_buf = _hash_getbuf_with_strategy(fc_rel, fc_blkno, HASH_READ, 0, fc_bstrategy);
	fc_page = BufferGetPage(fc_buf);

	if (PageGetSpecialSize(fc_page) == MAXALIGN(sizeof(HashPageOpaqueData)))
	{
		HashPageOpaque fc_opaque;

		fc_opaque = HashPageGetOpaque(fc_page);
		switch (fc_opaque->hasho_flag & LH_PAGE_TYPE)
		{
			case LH_UNUSED_PAGE:
				fc_stat->free_space += BLCKSZ;
				break;
			case LH_BUCKET_PAGE:
			case LH_OVERFLOW_PAGE:
				fc_pgstat_index_page(fc_stat, fc_page, FirstOffsetNumber,
								  PageGetMaxOffsetNumber(fc_page));
				break;
			case LH_BITMAP_PAGE:
			case LH_META_PAGE:
			default:
				break;
		}
	}
	else
	{
		/* 可能已损坏 */
	}

	_hash_relbuf(fc_rel, fc_buf);
}

/*
 * pgstat_gist_page -- 检查gist页面中的元组
 */
static void fc_pgstat_gist_page(pgstattuple_type *fc_stat, Relation fc_rel, BlockNumber fc_blkno,
				 BufferAccessStrategy fc_bstrategy)
{
	Buffer		fc_buf;
	Page		fc_page;

	fc_buf = ReadBufferExtended(fc_rel, MAIN_FORKNUM, fc_blkno, RBM_NORMAL, fc_bstrategy);
	LockBuffer(fc_buf, GIST_SHARE);
	gistcheckpage(fc_rel, fc_buf);
	fc_page = BufferGetPage(fc_buf);

	if (GistPageIsLeaf(fc_page))
	{
		fc_pgstat_index_page(fc_stat, fc_page, FirstOffsetNumber,
						  PageGetMaxOffsetNumber(fc_page));
	}
	else
	{
		/* 根节点或节点 */
	}

	UnlockReleaseBuffer(fc_buf);
}

/*
 * pgstat_index -- 返回通用索引中的活/死元组信息
 */
static Datum fc_pgstat_index(Relation fc_rel, BlockNumber fc_start, pgstat_page fc_pagefn,
			 FunctionCallInfo fc_fcinfo)
{
	BlockNumber fc_nblocks;
	BlockNumber fc_blkno;
	BufferAccessStrategy fc_bstrategy;
	pgstattuple_type fc_stat = {0};

	/* 为此索引准备访问策略 */
	fc_bstrategy = GetAccessStrategy(BAS_BULKREAD);

	fc_blkno = fc_start;
	for (;;)
	{
		/* 获取当前关系长度 */
		LockRelationForExtension(fc_rel, ExclusiveLock);
		fc_nblocks = RelationGetNumberOfBlocks(fc_rel);
		UnlockRelationForExtension(fc_rel, ExclusiveLock);

		/* 如果我们扫描了整个关系，则退出 */
		if (fc_blkno >= fc_nblocks)
		{
			fc_stat.table_len = (uint64) fc_nblocks * BLCKSZ;

			break;
		}

		for (; fc_blkno < fc_nblocks; fc_blkno++)
		{
			CHECK_FOR_INTERRUPTS();

			fc_pagefn(&fc_stat, fc_rel, fc_blkno, fc_bstrategy);
		}
	}

	relation_close(fc_rel, AccessShareLock);

	return fc_build_pgstattuple_type(&fc_stat, fc_fcinfo);
}

/*
 * pgstat_index_page -- 针对通用索引页面
 */
static void fc_pgstat_index_page(pgstattuple_type *fc_stat, Page fc_page,
				  OffsetNumber fc_minoff, OffsetNumber fc_maxoff)
{
	OffsetNumber fc_i;

	fc_stat->free_space += PageGetFreeSpace(fc_page);

	for (fc_i = fc_minoff; fc_i <= fc_maxoff; fc_i = OffsetNumberNext(fc_i))
	{
		ItemId		fc_itemid = PageGetItemId(fc_page, fc_i);

		if (ItemIdIsDead(fc_itemid))
		{
			fc_stat->dead_tuple_count++;
			fc_stat->dead_tuple_len += ItemIdGetLength(fc_itemid);
		}
		else
		{
			fc_stat->tuple_count++;
			fc_stat->tuple_len += ItemIdGetLength(fc_itemid);
		}
	}
}
